package com.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author Gerry chan
 * @version 1.0
 * 2020/12/27
 * DStream操作-无状态转换：https://www.bilibili.com/video/BV11A411L7CK?p=194
 */
object SparkStreaming06_state_Transform {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]")
      .setAppName("transform")

    val ssc = new StreamingContext(sparkConf, Seconds(3))

    val lines = ssc.socketTextStream("localhost", 999)

    //transform方法可以将底层RDD获取到后进行操作
    //Code:Driver端
    val newDS:DStream[String] = lines.transform(
      rdd => {
        //Code:Driver端 在此处写代码是在Driver端执行 （周期性执行）
        // 在算子之外在Driver端执行
        rdd.map(
          //Code:Executor 在此处写代码是在Executor端执行
          str => str
        )
      }
    )
    //Code:Driver
    val newDS1:DStream[String] = lines.map(
      data=>{
        //Code:Executor 在此处写代码是在Executor端执行
        data
      }
    )
  }
}
